package spring.cloud.tasks.api.service.impl;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.database.entity.NameSpaceDto;
import spring.cloud.tasks.api.database.entity.NameSpaceZooKeeperClusterMappingDto;
import spring.cloud.tasks.api.database.entity.ZooKeeperClusterDto;
import spring.cloud.tasks.api.database.service.NameSpaceDtoService;
import spring.cloud.tasks.api.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.api.database.service.ZooKeeperClusterDtoService;
import spring.cloud.tasks.api.domain.NameSpace;
import spring.cloud.tasks.api.domain.NameSpaceClient;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.ZooKeeperCluster;
import spring.cloud.tasks.api.service.ReportAlarmService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.ExecutorNodePath;
import spring.cloud.tasks.api.utils.SelfNodePath;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

import javax.annotation.Resource;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

public class ZooKeeperServiceImpl implements ZooKeeperService {


    protected static final String NAMESPACE_CREATOR_NAME = "REST_API";

    protected static final String ERR_MSG_TEMPLATE_FAIL_TO_CREATE = "Fail to create new namespace {%s} for reason {%s}";

    protected static final String ERR_MSG_NS_NOT_FOUND = "The namespace does not exists.";

    protected static final String ERR_MSG_NS_ALREADY_EXIST = "Invalid request. Namespace: {%s} already existed";

    private static final Logger log = LoggerFactory.getLogger(ZooKeeperServiceImpl.class);

    // namespace is unique in all zkClusters
    private final ConcurrentHashMap<String /** nns */, NameSpaceClient> nameSpaceToNameSpaceClientMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Object> nameSpaceLock = new ConcurrentHashMap<>(); // maybe could remove in right time
    private final Set<String> restrictComputeZkClusterKeys = Sets.newHashSet();
    @Resource
    private ZooKeeperRepository zooKeeperRepository;
    @Resource
    private ReportAlarmService reportAlarmService;
    @Resource
    private ZooKeeperClusterDtoService zooKeeperClusterDtoService;

    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;
    @Resource
    private NameSpaceDtoService nameSpaceDtoService;
    /**
     * 为保证values有序
     **/
    private final LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<>();
    private final List<String> onlineNameSpaceList = new ArrayList<>();


    /**
     * zooKeeperClusterKeyToZooKeeperClusterMap
     *
     * @param newZooKeeperClusterKeyToZooKeeperClusterMap
     */
    private void closeInvalidZooKeeperClusterClient(Map<String, ZooKeeperCluster> newZooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = this.zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ZooKeeperCluster> next = iterator.next();
            String zooKeeperClusterKey = next.getKey();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            //
            if (!newZooKeeperClusterKeyToZooKeeperClusterMap.containsKey(zooKeeperClusterKey)) {
                iterator.remove();
                closeZooKeeperClusterClient(zooKeeperCluster);
            } else {
                ZooKeeperCluster newZooKeeperCluster = newZooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
                if (zooKeeperCluster.equals(newZooKeeperCluster)) {
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else if (zooKeeperCluster.equalsNoNeedReconnect(newZooKeeperCluster)) {
                    zooKeeperCluster.setDescription(newZooKeeperCluster.getDescription());
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else {
                    iterator.remove();
                    closeZooKeeperClusterClient(zooKeeperCluster);
                }
            }
        }
    }

    private void connectZooKeeperCluster(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ZooKeeperCluster> next = iterator.next();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
            if (zooKeeper == null) {
                connectZooKeeperCluster(zooKeeperCluster);
                if (!zooKeeperCluster.isOffline()) {
                    //...
                }
            }
        }
    }

    private void initNameSpace(List<String> onlineNameSpaceList,//
                               ZooKeeperCluster zooKeeperCluster,//
                               List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList,//
                               List<NameSpaceConfiguration> nameSpaceConfigurationList) {//
        if (nameSpaceZooKeeperClusterMappingDtoList == null || zooKeeperCluster.isOffline()) {
            return;
        }
        ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
        List<NameSpaceConfiguration> newNameSpaceConfigurationList = new ArrayList<>();
        try {
            for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
                String nameSpace = dto.getNameSpace();
                // 过滤$SaturnSelf
                if (SelfNodePath.ROOT_NAME.equals(nameSpace)) {
                    continue;
                }
                // 如果这个标记为true，意味是新域，或者是迁移过来的域
                boolean isNamespaceNotIncludeInOriginRegCenerConfList = false;
                if (isNameSpaceNotIncludeInNameSpaceConfigurationList(nameSpace, nameSpaceConfigurationList)) {
                    // 对于新添加的域，需要初始化一些znode
                    initNameSpaceZooKeeperNodeIfNecessary(nameSpace, zooKeeper);
                    isNamespaceNotIncludeInOriginRegCenerConfList = true;
                }

                try {
                    if (isNamespaceNotIncludeInOriginRegCenerConfList) {
                        NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(zooKeeperCluster.getZooKeeperAddress(), nameSpace);
                        nameSpaceConfiguration.setZooKeeperClusterKey(zooKeeperCluster.getZooKeeperClusterKey());
                        nameSpaceConfiguration.setAlias(zooKeeperCluster.getAlias());
                        newNameSpaceConfigurationList.add(nameSpaceConfiguration);
                        if (!onlineNameSpaceList.contains(nameSpace)) {
                            onlineNameSpaceList.add(nameSpace);
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        int oldSize = nameSpaceConfigurationList != null ? nameSpaceConfigurationList.size() : 0;
        if (oldSize != newNameSpaceConfigurationList.size()) {
            log.info("Zkcluster [{}] namespace size change from {} to {}", zooKeeperCluster.getZooKeeperClusterKey(), oldSize, newNameSpaceConfigurationList.size());
        }

        zooKeeperCluster.setNameSpaceConfigurationList(newNameSpaceConfigurationList);
    }

    private boolean isNameSpaceNotIncludeInNameSpaceConfigurationList(String namespace, List<NameSpaceConfiguration> nameSpaceConfigurationList) {
        if (nameSpaceConfigurationList == null || nameSpaceConfigurationList.isEmpty()) {
            return true;
        }
        for (NameSpaceConfiguration nameSpaceConfiguration : nameSpaceConfigurationList) {
            if (namespace.equals(nameSpaceConfiguration.getNameSpace())) {
                return false;
            }
        }
        return true;
    }


    /**
     * 从数据库获取基础信息
     *
     * @return
     */
    private Map<String, ZooKeeperCluster> getZooKeeperClusterKeyToZooKeeperClusterMap() {
        LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<>();
        List<ZooKeeperClusterDto> dtoList = zooKeeperClusterDtoService.getZooKeeperClusterInfoList();
        if (dtoList != null) {
            for (ZooKeeperClusterDto dto : dtoList) {
                ZooKeeperCluster zooKeeperCluster = new ZooKeeperCluster();
                zooKeeperCluster.setZooKeeperClusterKey(dto.getZooKeeperClusterKey());
                zooKeeperCluster.setAlias(dto.getAlias());
                zooKeeperCluster.setZooKeeperAddress(dto.getConnectString());
                zooKeeperCluster.setDescription(dto.getDescription());
                zooKeeperClusterKeyToZooKeeperClusterMap.put(dto.getZooKeeperClusterKey(), zooKeeperCluster);
            }
        }
        return zooKeeperClusterKeyToZooKeeperClusterMap;
    }


    private Object getNameSpaceLock(String nameSpace) {
        Object lock = nameSpaceLock.get(nameSpace);
        if (lock == null) {
            lock = new Object();
            Object pre = nameSpaceLock.putIfAbsent(nameSpace, lock);
            if (pre != null) {
                lock = pre;
            }
        }
        return lock;
    }

    private void closeNameSpaceClient(String nameSpace) {
        try {
            NameSpaceClient nameSpaceClient = nameSpaceToNameSpaceClientMap.remove(nameSpace);
            if (nameSpaceClient != null) {
                nameSpaceClient.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            /*TaskManager taskAllotManager = nameSpaceToTaskManagerMap.remove(nameSpace);
            if (taskAllotManager != null) {
                taskAllotManager.stopWithCurator();
            }*/
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    /**
     * Close dashboardLeaderTreeCache, registryCenterClient, namespaceShardingListenerManager with this zkCluster
     */
    private void closeZooKeeperClusterClient(ZooKeeperCluster zooKeeperCluster) {
        try {
            List<NameSpaceConfiguration> namespaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            if (namespaceConfigurationList != null) {
                for (NameSpaceConfiguration nameSpaceConfiguration : namespaceConfigurationList) {
                    String nameSpace = nameSpaceConfiguration.getNameSpace();
                    synchronized (getNameSpaceLock(nameSpace)) {
                        closeNameSpaceClient(nameSpace);
                    }
                }
            }
            if (zooKeeperCluster.getConnectionStateListener() != null) {
                zooKeeperCluster.getConnectionStateListener().shutdownNowUntilTerminated();
                zooKeeperCluster.setConnectionStateListener(null);
            }
            if (zooKeeperCluster.getZooKeeper() != null && zooKeeperCluster.getZooKeeper().getCuratorFramework() != null) {
                zooKeeperCluster.getZooKeeper().getCuratorFramework().close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void connectZooKeeperCluster(final ZooKeeperCluster zooKeeperCluster) {
        String zooKeeperAddress = zooKeeperCluster.getZooKeeperAddress();
        try {
            final CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddress, null, zooKeeperCluster.getDigest());
            if (curatorFramework == null) {
                log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
                zooKeeperCluster.setZooKeeper(null);
                zooKeeperCluster.setConnectionStateListener(null);
                zooKeeperCluster.setOffline(true);
            } else {
                ConnectionStateListener connectionStateListener = new ConnectionStateListener("zk-connectionListener-thread-for-zkCluster-" + zooKeeperCluster.getAlias()) {
                    @Override
                    public void stop() {
                        zooKeeperCluster.setOffline(true);
                    }

                    @Override
                    public void restart() {
                        try {
                            zooKeeperCluster.setOffline(false);
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }
                };
                zooKeeperCluster.setZooKeeper(zooKeeperRepository.getZooKeeper(curatorFramework));
                zooKeeperCluster.setConnectionStateListener(connectionStateListener);
                zooKeeperCluster.setOffline(false);
                curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
            }
        } catch (Exception e) {
            log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
            log.error(e.getMessage(), e);
            zooKeeperCluster.setZooKeeper(null);
            zooKeeperCluster.setConnectionStateListener(null);
            zooKeeperCluster.setOffline(true);
        }
    }

    protected void initNameSpaceZooKeeperNodeIfNecessary(String nameSpace, ZooKeeper zooKeeper) {
        try {
            String taskIdsNodePath = "/" + nameSpace + TaskNodePath.getTasksNodePath();
            if (!zooKeeper.checkExists(taskIdsNodePath)) {
                zooKeeper.create(taskIdsNodePath);
            }
            String executorIdsNodePath = "/" + nameSpace + ExecutorNodePath.get$ExecutorNodePath();
            if (!zooKeeper.checkExists(executorIdsNodePath)) {
                zooKeeper.create(executorIdsNodePath);
            }

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    @Override
    public NameSpaceClient connectByNameSpace(String nameSpace) {
        NameSpaceConfiguration nameSpaceConfiguration = getNameSpaceConfigurationByNameSpace(nameSpace);
        if (nameSpaceConfiguration == null) {
            return new NameSpaceClient();
        }
        String zooKeeperAddressList = nameSpaceConfiguration.getZooKeeperAddressList();
        String digest = nameSpaceConfiguration.getDigest();
        synchronized (getNameSpaceLock(nameSpace)) {
            if (!nameSpaceToNameSpaceClientMap.containsKey(nameSpace)) {
                final NameSpaceClient nameSpaceClient = new NameSpaceClient();
                nameSpaceClient.setNameSpace(nameSpace);
                nameSpaceClient.setZooKeeperAddress(zooKeeperAddressList);
                CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddressList, nameSpace, digest);
                if (curatorFramework == null) {
                    return nameSpaceClient;
                }
                nameSpaceClient.setConnected(curatorFramework.getZookeeperClient().isConnected());
                nameSpaceClient.setCuratorFramework(curatorFramework);
                nameSpaceToNameSpaceClientMap.put(nameSpace, nameSpaceClient);
                return nameSpaceClient;
            } else {
                NameSpaceClient nameSpaceClient = nameSpaceToNameSpaceClientMap.get(nameSpace);
                if (nameSpaceClient == null) {
                    nameSpaceClient = new NameSpaceClient();
                    nameSpaceClient.setNameSpace(nameSpace);
                    nameSpaceClient.setZooKeeperAddress(zooKeeperAddressList);
                } else {
                    if (nameSpaceClient.getCuratorFramework() != null) {
                        nameSpaceClient.setConnected(nameSpaceClient.getCuratorFramework().getZookeeperClient().isConnected());
                    } else {
                        nameSpaceClient.setConnected(false);
                    }
                }
                return nameSpaceClient;
            }
        }
    }


    @Override
    public NameSpaceConfiguration getNameSpaceConfigurationByNameSpace(String nameSpace) {
        if (Strings.isNullOrEmpty(nameSpace)) {
            return null;
        }
        Collection<ZooKeeperCluster> zooKeeperClusterCollection = zooKeeperClusterKeyToZooKeeperClusterMap.values();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterCollection) {
            for (NameSpaceConfiguration nameSpaceConfiguration : zooKeeperCluster.getNameSpaceConfigurationList()) {
                if (nameSpaceConfiguration != null && nameSpace.equals(nameSpaceConfiguration.getNameSpace())) {
                    return nameSpaceConfiguration;
                }
            }
        }
        return null;
    }

    @Override
    public List<NameSpaceConfiguration> findConfigsByZkCluster(ZooKeeperCluster zooKeeperCluster) {
        List<String> nameSpaceList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperCluster.getZooKeeperClusterKey());
        if (nameSpaceList == null || nameSpaceList.isEmpty()) {
            return Lists.newArrayList();
        }

        List<NameSpaceConfiguration> nameSpaceConfigurationList = Lists.newArrayList();
        for (String nameSpace : nameSpaceList) {
            NameSpaceConfiguration nameSpaceConfiguration = constructRegistryCenterConfiguration(zooKeeperCluster, nameSpace);
            nameSpaceConfigurationList.add(nameSpaceConfiguration);
        }

        return nameSpaceConfigurationList;
    }

    protected NameSpaceConfiguration constructRegistryCenterConfiguration(ZooKeeperCluster zooKeeperCluster, String namespace) {
        NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(zooKeeperCluster.getZooKeeperAddress(), namespace);
        nameSpaceConfiguration.setZooKeeperClusterKey(zooKeeperCluster.getZooKeeperClusterKey());
        nameSpaceConfiguration.setAlias(zooKeeperCluster.getAlias());
        return nameSpaceConfiguration;
    }

    @Override
    public ZooKeeper connectOnly(String connectString, String namespace) throws RuntimeException {
        CuratorFramework curatorFramework = zooKeeperRepository.connect(connectString, namespace, null);
        if (curatorFramework != null) {
            return zooKeeperRepository.getZooKeeper(curatorFramework);
        }
        return null;
    }

    @Override
    public void notifyRefresh() throws RuntimeException {
        //DB更新 UUID
    }


    @Override
    public ZooKeeperCluster getZooKeeperCluster(String zooKeeperClusterKey) {
        return zooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void createZooKeeperClusterInfo(String zooKeeperClusterKey, String alias, String connectString, String description) throws RuntimeException {
        ZooKeeperClusterDto zooKeeperClusterInfo = zooKeeperClusterDtoService.getZooKeeperClusterInfoByZooKeeperClusterKey(zooKeeperClusterKey);
        if (zooKeeperClusterInfo != null) {
            throw new RuntimeException(String.format("ZK cluster[%s]已经存在", zooKeeperClusterKey));
        }
        zooKeeperClusterDtoService.createZooKeeperClusterInfo(zooKeeperClusterKey, alias, connectString, description, "");
        notifyRefresh();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateZkCluster(String zooKeeperClusterKey, String connectString, String description) throws RuntimeException {
        ZooKeeperClusterDto zooKeeperClusterInfo = zooKeeperClusterDtoService.getZooKeeperClusterInfoByZooKeeperClusterKey(zooKeeperClusterKey);
        if (zooKeeperClusterInfo == null) {
            throw new RuntimeException(String.format("ZK cluster[%s]不存在", zooKeeperClusterKey));
        }
        // cannot change alias but connectString and description
        zooKeeperClusterInfo.setConnectString(connectString);
        zooKeeperClusterInfo.setDescription(description);
        zooKeeperClusterInfo.setLastUpdateTime(new Date());
        zooKeeperClusterDtoService.updateZooKeeperClusterInfo(zooKeeperClusterInfo);
        notifyRefresh();
    }

    @Override
    public List<ZooKeeperCluster> getZooKeeperClusterList() {
        return new ArrayList<>(zooKeeperClusterKeyToZooKeeperClusterMap.values());
    }

    @Override
    public List<ZooKeeperCluster> getOnlineZooKeeperClusterList() {
        List<ZooKeeperCluster> onlineZooKeeperClusterList = Lists.newArrayList();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterKeyToZooKeeperClusterMap.values()) {
            if (!zooKeeperCluster.isOffline()) {
                onlineZooKeeperClusterList.add(zooKeeperCluster);
            }
        }
        return onlineZooKeeperClusterList;
    }

    @Override
    public int domainCount(String zooKeeperClusterKey) {
        ZooKeeperCluster zooKeeperCluster = zooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
        if (zooKeeperCluster != null) {
            List<NameSpaceConfiguration> nameSpaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            if (nameSpaceConfigurationList != null) {
                return nameSpaceConfigurationList.size();
            }
        }
        return 0;
    }

    @Override
    public boolean nameSpaceIsCorrect(String nameSpace, CuratorFramework curatorFramework) throws RuntimeException {
        if (SelfNodePath.ROOT_NAME.equals(nameSpace)) {
            return false;
        }
        try {
            String executorsPath = "/" + nameSpace + ExecutorNodePath.getExecutorNodePath();
            if (curatorFramework.checkExists().forPath(executorsPath) != null) {
                return true;
            }
            String jobsPath = "/" + nameSpace + TaskNodePath.getTasksNodePath();
            return curatorFramework.checkExists().forPath(jobsPath) != null;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return false;
        }
    }

    @Override
    public List<String> getOnlineNameSpaceList() throws RuntimeException {
        return onlineNameSpaceList;
    }

    @Transactional(rollbackFor = {Exception.class})
    @Override
    public void createNameSpace(NameSpace nameSpace) throws RuntimeException {
        String nameSpaceString = nameSpace.getNameSpace();
        String zooKeeperClusterKey = nameSpace.getZooKeeperClusterKey();
        ZooKeeperCluster zooKeeperCluster = getZooKeeperCluster(zooKeeperClusterKey);
        if (zooKeeperCluster == null) {
            throw new RuntimeException(HttpStatus.BAD_REQUEST.value() + String.format(ERR_MSG_TEMPLATE_FAIL_TO_CREATE, nameSpace, "not found zkcluster" + zooKeeperClusterKey));
        }
        if (checkNamespaceExists(nameSpaceString)) {
            throw new RuntimeException(HttpStatus.BAD_REQUEST.value() + String.format(ERR_MSG_NS_ALREADY_EXIST, nameSpace));
        }
        try {
            // 创建 namespaceInfo
            NameSpaceDto nameSpaceDto = constructNamespaceInfo(nameSpace);
            nameSpaceDtoService.create(nameSpaceDto);
            // 创建 zkcluster 和 namespaceInfo 关系
            nameSpaceZooKeeperClusterMappingDtoService.insert(nameSpaceString, zooKeeperClusterKey, NAMESPACE_CREATOR_NAME);
            // refresh
            notifyRefresh();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(HttpStatus.INTERNAL_SERVER_ERROR.value() + String.format(ERR_MSG_TEMPLATE_FAIL_TO_CREATE, nameSpace, e.getMessage()));
        }
    }


    @Override
    public NameSpace getNameSpace(String nameSpace) throws RuntimeException {
        if (nameSpaceDtoService.selectByNameSpace(nameSpace) == null) {
            throw new RuntimeException(HttpStatus.NOT_FOUND.value() + ERR_MSG_NS_NOT_FOUND);
        }

        String zooKeeperClusterKey = nameSpaceZooKeeperClusterMappingDtoService.getZooKeeperClusterKey(nameSpace);
        if (StringUtils.isBlank(zooKeeperClusterKey)) {
            throw new RuntimeException(HttpStatus.NOT_FOUND.value() + ERR_MSG_NS_NOT_FOUND);
        }

        NameSpace nameSpaceInfoVo = new NameSpace();
        nameSpaceInfoVo.setNameSpace(nameSpace);
        nameSpaceInfoVo.setZooKeeperClusterKey(zooKeeperClusterKey);

        return nameSpaceInfoVo;
    }


    private boolean checkNamespaceExists(String namespace) {
        if (nameSpaceDtoService.selectByNameSpace(namespace) != null) {
            return true;
        }
        // 判断其它集群是否有该域
        String zooKeeperClusterKeyOther = nameSpaceZooKeeperClusterMappingDtoService.getZooKeeperClusterKey(namespace);
        return zooKeeperClusterKeyOther != null;
    }

    private NameSpaceDto constructNamespaceInfo(NameSpace nameSpace) {
        NameSpaceDto nameSpaceDto = new NameSpaceDto();
        nameSpaceDto.setCreatedBy(NAMESPACE_CREATOR_NAME);
        nameSpaceDto.setCreateTime(new Date());
        nameSpaceDto.setLastUpdatedBy(NAMESPACE_CREATOR_NAME);
        nameSpaceDto.setLastUpdateTime(new Date());
        //
        nameSpaceDto.setNamespace(nameSpace.getNameSpace());
        nameSpaceDto.setContent(nameSpace.getContent());
        nameSpaceDto.setIsDeleted(0);
        nameSpaceDto.setBusId("");
        return nameSpaceDto;
    }

    @Override
    public ZooKeeper getZooKeeper(String nameSpace) throws RuntimeException {
        ZooKeeper zooKeeper = null;
        try {
            NameSpaceConfiguration nameSpaceConfiguration = getNameSpaceConfigurationByNameSpace(nameSpace);
            if (nameSpaceConfiguration == null) {
                throw new RuntimeException("find registryCenterConfiguration failed");
            }
            String zooKeeperAddressList = nameSpaceConfiguration.getZooKeeperAddressList();
            String digest = nameSpaceConfiguration.getDigest();
            synchronized (getNameSpaceLock(nameSpace)) {
                if (!nameSpaceToNameSpaceClientMap.containsKey(nameSpace)) {
                    final NameSpaceClient nameSpaceClient = new NameSpaceClient();
                    nameSpaceClient.setNameSpace(nameSpace);
                    nameSpaceClient.setZooKeeperAddress(zooKeeperAddressList);
                    CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddressList, nameSpace, digest);
                    if (curatorFramework != null) {
                        nameSpaceClient.setConnected(curatorFramework.getZookeeperClient().isConnected());
                        nameSpaceClient.setCuratorFramework(curatorFramework);
                        nameSpaceToNameSpaceClientMap.put(nameSpace, nameSpaceClient);
                        zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                    }
                } else {
                    NameSpaceClient nameSpaceClient = nameSpaceToNameSpaceClientMap.get(nameSpace);
                    if (nameSpaceClient != null) {
                        CuratorFramework curatorFramework = nameSpaceClient.getCuratorFramework();
                        if (curatorFramework != null) {
                            nameSpaceClient.setConnected(curatorFramework.getZookeeperClient().isConnected());
                            zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                        }
                    }
                }
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        if (zooKeeper == null) {
            throw new RuntimeException("Connect zookeeper failed");
        }
        return zooKeeper;
    }

    @Override
    public synchronized void refreshRegistryCenterForNamespace(String zooKeeperClusterKey, String nameSpace) {

        Map<String, ZooKeeperCluster> targetZooKeeperClusterKeyToZooKeeperClusterMap = getTargetZooKeeperClusterKeyToZooKeeperClusterMap(zooKeeperClusterKey);
        updateRegistryCenterConfiguration(nameSpace, targetZooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey).getZooKeeperAddress(), zooKeeperClusterKey);

        closeInvalidZooKeeperClusterClient(targetZooKeeperClusterKeyToZooKeeperClusterMap);
        connectZooKeeperCluster(targetZooKeeperClusterKeyToZooKeeperClusterMap);

        ZooKeeperCluster zooKeeperCluster = targetZooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
        if (zooKeeperCluster != null) {
            List<String> allOnlineNamespacesTemp = new ArrayList<>();
            List filteredList = getTargetNamespaceZkClusterMapping(zooKeeperClusterKey, nameSpace);
            initNameSpace(allOnlineNamespacesTemp, zooKeeperCluster, filteredList, zooKeeperCluster.getNameSpaceConfigurationList());
            updateAllOnlineNamespaces(nameSpace);
        }
        log.info("refreshRegistryCenterForNamespace done : {}, {}", zooKeeperClusterKey, nameSpace);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void deleteZooKeeperClusterInfo(String zooKeeperClusterKey) throws RuntimeException {
        if (getZooKeeperCluster(zooKeeperClusterKey) == null) {
            throw new RuntimeException("fail to delete.for ZkCluster does not exist");
        }
        if (domainCount(zooKeeperClusterKey) > 0) {
            throw new RuntimeException("fail to delete.for ZkCluster still has domains");
        }
        if (!CollectionUtils.isEmpty(nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperClusterKey))) {
            throw new RuntimeException("fail to delete.for ZkCluster still has domains");
        }
        zooKeeperClusterDtoService.deleteZooKeeperClusterInfo(zooKeeperClusterKey);
        notifyRefresh();
    }

    private Map<String, ZooKeeperCluster> getTargetZooKeeperClusterKeyToZooKeeperClusterMap(String zooKeeperClusterKey) {
        Map<String, ZooKeeperCluster> result = new HashMap<>(1);
        Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = getZooKeeperClusterKeyToZooKeeperClusterMap();
        Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ZooKeeperCluster> entry = iterator.next();
            if (entry.getKey().equals(zooKeeperClusterKey)) {
                result.put(zooKeeperClusterKey, entry.getValue());
                break;
            }
        }
        return result;
    }


    private List<NameSpaceZooKeeperClusterMappingDto> getTargetNamespaceZkClusterMapping(String zooKeeperClusterKey, String namespace) {
        List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceZooKeeperClusterMappingDtoList(zooKeeperClusterKey);
        List<NameSpaceZooKeeperClusterMappingDto> filteredList = new ArrayList<NameSpaceZooKeeperClusterMappingDto>();
        for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
            if (dto.getNameSpace().equals(namespace)) {
                filteredList.add(dto);
                break;
            }
        }
        return nameSpaceZooKeeperClusterMappingDtoList;
    }

    private void updateRegistryCenterConfiguration(String nameSpace, String zkAddressList, String zkClusterName) {
        List<NameSpaceConfiguration> sourceRegCenterConfList = zooKeeperClusterKeyToZooKeeperClusterMap.get(zkClusterName).getNameSpaceConfigurationList();
        boolean regCenterConfAlreadyExisted = false;
        for (NameSpaceConfiguration nameSpaceConfiguration : sourceRegCenterConfList) {
            if (nameSpaceConfiguration.getNameSpace().equals(nameSpace)) {
                regCenterConfAlreadyExisted = true;
                break;
            }
        }
        if (!regCenterConfAlreadyExisted) {
            NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(zkAddressList, nameSpace);
            zooKeeperClusterKeyToZooKeeperClusterMap.get(zkClusterName).getNameSpaceConfigurationList().add(nameSpaceConfiguration);
        }
    }

    private void updateAllOnlineNamespaces(String nameSpace) {
        if (!onlineNameSpaceList.contains(nameSpace)) {
            onlineNameSpaceList.add(nameSpace);
        }
    }
}
